package io.github.viscent.mtpattern.ch8.activeobject.example;

import org.apache.log4j.Logger;

import java.io.IOException;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 彩信下发请求缓存 模式角色：ActiveObject.Proxy
 */
@SuppressWarnings("all")
public class AsyncRequestPersistence implements RequestPersistence {

    final static Logger logger = Logger.getLogger(AsyncRequestPersistence.class);

    /**
     * 1分钟
     */
    private static final long ONE_MINUTE_IN_SECONDS = 60;

    /**
     * 一段时间区间任务消耗时间
     */
    final AtomicLong taskTimeConsumedPerInterval = new AtomicLong(0);

    /**
     * 一段时间区间任务提交数量
     */
    final AtomicInteger requestSubmittedPerInterval = new AtomicInteger(0);

    /**
     * 模式角色：ActiveObject.Servant
     */
    private final DiskbasedRequestPersistence delegate = new DiskbasedRequestPersistence();

    /**
     * 模式角色：ActiveObject.Scheduler
     */
    private final ThreadPoolExecutor scheduler;

    /**
     * 用于保存AsyncRequestPersistence的唯一实例
     */
    private static class InstanceHolder {
        final static RequestPersistence INSTANCE = new AsyncRequestPersistence();
    }

    private AsyncRequestPersistence() {
        ThreadFactory threadFactory = r -> new Thread(r, "AsyncRequestPersistence");

        // ArrayBlockingQueue相当于模式角色：ActiveObject.ActivationQueue
        scheduler = new ThreadPoolExecutor(1, 3, 60 * ONE_MINUTE_IN_SECONDS, TimeUnit.SECONDS,
            new ArrayBlockingQueue<Runnable>(200), threadFactory, new ThreadPoolExecutor.CallerRunsPolicy());

        // 启动队列监控定时任务
        TimerTask timerTask = new TimerTask() {
            @Override
            public void run() {
                if (logger.isInfoEnabled()) {
                    logger.info("task count:" + requestSubmittedPerInterval + ",Queue size:" + scheduler.getQueue().size() + ",taskTimeConsumedPerInterval:" + taskTimeConsumedPerInterval.get() + " ms");
                }
                taskTimeConsumedPerInterval.set(0);
                requestSubmittedPerInterval.set(0);
            }
        };
        new Timer(true).scheduleAtFixedRate(timerTask, 0, ONE_MINUTE_IN_SECONDS * 1000);
    }

    /**
     * 获取类AsyncRequestPersistence的唯一实例
     */
    public static RequestPersistence getInstance() {
        return InstanceHolder.INSTANCE;
    }

    /**
     * 彩信下发请求缓存
     */
    @Override
    public void store(final MMSDeliverRequest request) {
        // 模式角色：ActiveObject.MethodRequest
        Callable<Boolean> methodRequest = () -> {
            long start = System.currentTimeMillis();
            try {
                delegate.store(request);
            } finally {
                taskTimeConsumedPerInterval.addAndGet(System.currentTimeMillis() - start);
            }
            return Boolean.TRUE;
        };
        scheduler.submit(methodRequest);
        requestSubmittedPerInterval.incrementAndGet();
    }

    @Override
    public void close() throws IOException {
        scheduler.shutdown();
    }

}